package com.qupeng.demo.kafka.kafkaapache.security;

import org.apache.kafka.common.security.plain.internals.PlainServerCallbackHandler;

import javax.security.auth.login.AppConfigurationEntry;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;

public class PasswordVerifier extends PlainServerCallbackHandler {
    private final List<String> passwdFiles = new ArrayList<>();

    public void configure(Map<String, ?> configs, String mechanism,
                          List<AppConfigurationEntry> jaasEntries) {
        Map<String, ?> loginOptions = jaasEntries.get(0).getOptions();
        String files = (String) loginOptions.get("password.files");
        Collections.addAll(passwdFiles, files.split(","));
    }

    @Override
    protected boolean authenticate(String user, char[] password) {
        return passwdFiles.stream()
                .anyMatch(file -> authenticate(file, user, password));
    }

    private boolean authenticate(String file, String user, char[] password) {
        try {
            String cmd = String.format("htpasswd -vb %s %s %s",
                    file, user, new String(password));
            return Runtime.getRuntime().exec(cmd).waitFor() == 0;
        } catch (Exception e) {
            return false;
        }
    }
}